package com.snail.starter.kafka;

import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import javax.annotation.Resource;
import java.time.temporal.ChronoUnit;

/**
 * @author: snail framework
 * @Date: 2023/1/31 下午5:21
 * @Version: 1.0.0
 */
@Configuration
public class KafkaConfig {

    @Resource
    KafkaProperties kafkaProperties;

    @Bean
    ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(kafkaProperties.buildAdminProperties());
    }

    @Bean
    ConcurrentKafkaListenerContainerFactory<String, String> batchKafkaMessageListenerContainer(ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        //消费工厂
        factory.setConsumerFactory(consumerFactory);
        //并行数
        factory.setConcurrency(kafkaProperties.getListener().getConcurrency());
        //批次监听
        factory.setBatchListener(true);
        //轮询超时时间
        factory.getContainerProperties().setPollTimeout(kafkaProperties.getListener().getPollTimeout().get(ChronoUnit.MILLIS));
        return factory;
    }

}